package mapred.config;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.mapred.OutputFormat;

@SuppressWarnings("unchecked")
public class OutputConfig {

    private Class<? extends OutputFormat> outputFormat;

    private Class<? extends Writable> outputKeyClass;

    private Class<? extends Writable> outputValueClass;

    private String outputPath;

    private Compress compress;

    private OutputConfig(Class<? extends OutputFormat> outputFormat,
            Class<? extends Writable> outputKeyClass,
            Class<? extends Writable> outputValueClass, String outputPath,
            Compress compress) {
        this.outputFormat = outputFormat;
        this.outputKeyClass = outputKeyClass;
        this.outputValueClass = outputValueClass;
        this.outputPath = outputPath;
        this.compress = (compress == null ? Compress.NULL : compress);
    }

    public static OutputConfig newInstance(
            Class<? extends OutputFormat> outputFormat,
            Class<? extends Writable> outputKeyClass,
            Class<? extends Writable> outputValueClass, String outputPath,
            Compress compress) {
        return new OutputConfig(outputFormat, outputKeyClass, outputValueClass,
                outputPath, compress);
    }

    public Class<? extends OutputFormat> getOutputFormat() {
        return outputFormat;
    }

    public Class<? extends Writable> getOutputKeyClass() {
        return outputKeyClass;
    }

    public Class<? extends Writable> getOutputValueClass() {
        return outputValueClass;
    }

    public String getOutputPath() {
        return outputPath;
    }

    public boolean isCompressOutput() {
        return (compress != Compress.NULL);
    }

    public Class<? extends CompressionCodec> getOutputCompressorClass() {
        return compress.getCompressionCodecClass();
    }

    public void replace(AppConfig app) {
        if (app.contains("-output")) {
            outputPath = app.get("-output");
        }
        if (app.contains("-compress")) {
            compress = Compress.valueOf(app.get("-compress").toUpperCase());
        }
    }

}
